package com.bangbang.mq;


import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.math.BigDecimal;
import java.util.Date;
import java.util.List;

@Component
@RocketMQMessageListener(consumerGroup = "orderGroup",topic = "order-scope")
public class MQconsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;


    //广播消息不支持
    @Override
    public void onMessage(String message) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                MessageExt message = list.get(0);
                try {
                    String msgId = message.getMsgId();
                    Object mQconsumer = redisTemplate.opsForHash().get("MQconsumer", msgId);
                    if (mQconsumer == null){

                        byte[] body = message.getBody();
                        String messageBody = new String(body, "utf-8");
                        System.err.println("消息"+messageBody+"Receive New Message"+ list);
                        /*TbUserScoreAccount parse = JSON.parseObject(messageBody,TbUserScoreAccount.class);
                        System.err.println(parse.getUserId());
                        System.err.println(parse.getTotalScore());
                        TbUserScoreAccount account = scoreServiceI.selectUserScore(parse.getUserId());
                        if(account == null){
                            parse.setUpdateTime(new Date());
                            parse.setCreatTime(new Date());
                            scoreServiceI.insertSelective(parse);
                        }
                        BigDecimal decimal = account.getTotalScore().add(parse.getTotalScore());
                        account.setUpdateTime(new Date());
                        account.setTotalScore(decimal);
                        scoreServiceI.updateUserScore(account);*/
                        //重试消息的ID
                        //消费过后，将消息存入redis
                        redisTemplate.opsForHash().put("MQconsumer",msgId,"1");
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                }catch (Exception e){
                    e.printStackTrace();
                    //消息重试次数
                    int rt = message.getReconsumeTimes();
                    if(rt>3){
                        //可以将该消息存入DB或Log日志中，或其他处理方式 进行重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                    System.err.println("进入死信消息");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }
}
